/**
 * Copyright 1999-2015 dangdang.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * </p>
 */
package spring.cloud.tasks.executor.node;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.zookeeper.data.Stat;
import spring.cloud.tasks.executor.exception.JobException;
import spring.cloud.tasks.executor.listener.config.TaskConfiguration;
import spring.cloud.tasks.executor.listener.task_execute.ExecutorNode;
import spring.cloud.tasks.executor.utils.SleepUtils;
import spring.cloud.tasks.executor.utils.execute_callback.LeaderExecuteCallback;
import spring.cloud.tasks.executor.utils.log.LogUtils;
import spring.cloud.tasks.executor.utils.zookeeper.NameSpaceConfiguration;
import spring.cloud.tasks.executor.utils.zookeeper.Zookeeper;
import spring.cloud.tasks.executor.utils.zookeeper.ZookeeperImpl;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 作业节点数据访问类.
 *
 * <p>
 * 作业节点是在普通的节点前加上作业名称的前缀.
 * </p>
 */
@Slf4j
@Setter
@Getter
public class TaskNodeStorage {

    private static final int MAX_DELETE_RETRY_TIMES = 10;
    private final Zookeeper zookeeper;
    private final String taskId;
    private final TaskConfiguration taskConfiguration;
    private String executorId;


    public TaskNodeStorage(final Zookeeper zookeeper, final TaskConfiguration taskConfiguration) {
        this.zookeeper = zookeeper;
        this.taskId = taskConfiguration.getTaskId();
        this.taskConfiguration = taskConfiguration;
        if (zookeeper != null) {
            executorId = zookeeper.getExecutorId();
        }
    }

    /**
     * 判断作业节点是否存在.
     *
     * @param node 作业节点名称
     * @return 作业节点是否存在
     */
    public boolean isTaskNodeExisted(final String node) {
        return zookeeper.checkExists(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node));
    }

    /**
     * 判断作业是否存在.
     *
     * @param taskId 作业节点名称
     * @return 作业是否存在
     */
    public boolean isTaskExisted(final String taskId) {
        return zookeeper.checkExists(TaskNodePaths.getTaskPath(taskId));
    }

    /**
     * 获取作业节点数据.
     *
     * @param node 作业节点名称
     * @return 作业节点数据值
     */
    public String getTaskNodeData(final String node) {
        return zookeeper.getData(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node));
    }

    /**
     * 获取作业节点子节点名称列表.
     *
     * @param node 作业节点名称
     * @return 作业节点子节点名称列表
     */
    public List<String> getTaskNodeChildrenKeys(final String node) {
        return zookeeper.getChildrenList(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node));
    }

    /**
     * 如果不存在则创建作业节点.
     *
     * @param node 作业节点名称
     */
    public void createTaskNodeIfNeeded(final String node) {
        zookeeper.persist(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node), "");
    }

    public TaskConfiguration getTaskConfiguration() {
        return taskConfiguration;
    }

    public void createOrUpdateTaskNodeWithValue(final String node, final String value) {
        zookeeper.persist(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node), value);
    }

    /**
     * 如果节点存在，则删除作业节点
     *
     * @param node 作业节点名称
     */
    public void removeTaskNodeIfExisted(final String node) {
        if (isTaskNodeExisted(node)) {
            zookeeper.delete(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node));
        }
    }

    /**
     * 删除作业节点
     *
     * @param node 作业节点名称
     */
    public void removeTaskNode(final String node) {
        zookeeper.delete(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node));
    }


    /**
     * 如果节点不存在或允许覆盖则填充节点数据.
     *
     * @param node  作业节点名称
     * @param value 作业节点数据值
     */
    public void fillTaskNodeIfNullOrOverwrite(final String node, final Object value) {
        if (null == value) {
            LogUtils.info(log, taskId, "job node value is null, node:{}", node);
            return;
        }
        if (!isTaskNodeExisted(node) || (!value.toString().equals(getTaskNodeData(node)))) {
            zookeeper.persist(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node), value.toString());
        }
    }

    /**
     * 填充临时节点数据.
     *
     * @param node  作业节点名称
     * @param value 作业节点数据值
     */
    public void fillEphemeralTaskNode(final String node, final Object value) {
        zookeeper.persistEphemeralDataNode(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node), value.toString());
    }

    /**
     * 更新节点数据.
     *
     * @param node  作业节点名称
     * @param value 作业节点数据值
     */
    public void updateTaskNode(final String node, final Object value) {
        zookeeper.setData(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node), value.toString());
    }

    /**
     * 跟新作业节点数据。可用于同一个namespace下的其他作业。
     *
     * @param jobName 作业名
     * @param node    作业节点名称
     * @param value   待替换的数据
     */
    public void updateTaskNode(final String jobName, final String node, final Object value) {
        zookeeper.setData(TaskNodePaths.getTaskNodePath(jobName, node), value.toString());
    }

    /**
     * 替换作业节点数据.
     *
     * @param node  作业节点名称
     * @param value 待替换的数据
     */
    public void replaceTaskNode(final String node, final Object value) {
        zookeeper.persist(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), node), value.toString());
    }

    /**
     * 替换作业节点数据.
     *
     * @param jobName 作业名
     * @param node    作业节点名称
     * @param value   待替换的数据
     */
    public void replaceTaskNode(final String jobName, final String node, final Object value) {
        zookeeper.persist(TaskNodePaths.getTaskNodePath(jobName, node), value.toString());
    }


    /**
     * 在主节点执行操作.
     *
     * @param latchNode               分布式锁使用的作业节点名称
     * @param leaderExecutionCallback 执行操作的回调
     */
    public void executeInLeader(final String latchNode, final LeaderExecuteCallback leaderExecutionCallback) {
        try (LeaderLatch leaderLatch = new LeaderLatch(getCuratorFramework(), TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), latchNode))) {
            leaderLatch.start();
            leaderLatch.await();
            leaderExecutionCallback.execute();
            // CHECKSTYLE:OFF
        } catch (final Exception e) {
            LogUtils.error(log, taskId, e.getMessage(), e);
            // CHECKSTYLE:ON
            if (e instanceof InterruptedException) {// NOSONAR
                Thread.currentThread().interrupt();
            } else {
                throw new JobException(e);
            }
        }
    }

    public void executeInLeader(final String latchNode, final LeaderExecuteCallback leaderExecutionCallback, final long timeout, final TimeUnit timeUnit, final LeaderExecuteCallback timeoutLeaderExecutionCallback) {
        try (LeaderLatch leaderLatch = new LeaderLatch(getCuratorFramework(), TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), latchNode))) {
            leaderLatch.start();
            if (leaderLatch.await(timeout, timeUnit)) {
                leaderExecutionCallback.execute();
            } else {
                if (timeoutLeaderExecutionCallback != null) {
                    timeoutLeaderExecutionCallback.execute();
                }
            }
            // CHECKSTYLE:OFF
        } catch (final Exception e) {
            LogUtils.error(log, taskId, e.getMessage(), e);
            // CHECKSTYLE:ON
            if (e instanceof InterruptedException) {// NOSONAR
                Thread.currentThread().interrupt();
            } else {
                throw new JobException(e);
            }
        }
    }

    public CuratorFramework getCuratorFramework() {
        return zookeeper.getCuratorFramework();
    }

    /**
     * 获取当前运行execution分片列表
     *
     * @return 当前运行execution分片列表
     */
    public List<String> getRunningItems() {
        return zookeeper.getChildrenList(TaskNodePaths.getTaskNodePath(taskConfiguration.getTaskId(), "execution"));
    }

    /**
     * 删除ZK结点
     */
    public void deleteJobNode() {
        NameSpaceConfiguration namespaceConfiguration = ((ZookeeperImpl) zookeeper).getNamespaceConfiguration();
        ZookeeperImpl zookeeper = new ZookeeperImpl(namespaceConfiguration);
        try {
            zookeeper.init(); // maybe throw RuntimeException
            zookeeper.delete(ExecutorNode.getExecutorNode(taskId, executorId));

            String fullPath = TaskNodePaths.getTaskPath(taskConfiguration.getTaskId());
            Stat stat = zookeeper.getStat(fullPath);
            if (stat == null) {
                return;
            }
            long ctime = stat.getCtime();
            for (int i = 0; i < MAX_DELETE_RETRY_TIMES; i++) {
                stat = zookeeper.getStat(fullPath);
                if (stat == null) {
                    return;
                }
                if (stat.getCtime() != ctime) {
                    LogUtils.info(log, taskId, "the job node ctime has changed, give up to delete");
                    return;
                }
                List<String> servers = zookeeper.getChildrenList(ExecutorNode.getExecutorRoot(taskId));
                if (servers == null || servers.isEmpty()) {
                    if (tryToRemoveNode(zookeeper, fullPath)) {
                        return;
                    }
                }
                SleepUtils.sleepOneSeconds();
            }
        } catch (Throwable t) {
            LogUtils.error(log, taskId, "delete job node error", t);
        } finally {
            zookeeper.close();
        }
    }

    private boolean tryToRemoveNode(ZookeeperImpl zookeeperRegistryCenter, String fullPath) {
        try {
            zookeeperRegistryCenter.delete(fullPath);
            return true;
        } catch (Exception e) {
            LogUtils.error(log, taskId, e.getMessage(), e);
        }
        return false;
    }

    public boolean isConnected() {
        return zookeeper.isConnected();
    }
}
